package com.ssdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.ssdp.dto.Result;
import com.ssdp.entity.SeckillVoucher;
import com.ssdp.entity.VoucherOrder;
import com.ssdp.mapper.SeckillVoucherMapper;
import com.ssdp.mapper.VoucherOrderMapper;
import com.ssdp.service.ISeckillVoucherService;
import com.ssdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.ssdp.utils.RedisIdWorker;
import com.ssdp.utils.SimpleRedisLock;
import com.ssdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.PostConstruct;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author syo 2319615164@qq.com
 * @description 优惠券订单服务
 * @date 2024/3/14
 * @Copyright syo
 */
@Service
@Slf4j
@SuppressWarnings("all")
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    /**
     * 秒杀券DAO
     */
    @Autowired
    private SeckillVoucherMapper seckillVoucherMapper;

    /**
     * 秒杀券服务
     */
    @Autowired
    private ISeckillVoucherService seckillVoucherService;

    /**
     * 优惠券订单DAO
     */
    @Autowired
    private VoucherOrderMapper voucherOrderMapper;

    /**
     * Redis ID生成器
     */
    @Autowired
    private RedisIdWorker redisIdWorker;

    /**
     * Redis 客户端
     */
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /**
     * Redisson客户端
     */
    @Autowired
    private RedissonClient redissonClient;

    /**
     * Redis脚本
     */
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

    /**
     * 类链接的解析阶段会执行
     */
    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    /**
     * 阻塞队列（第一种异步秒杀的消息队列）
     */
    private final BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);

    /**
     * 单线程的线程池
     */
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    /**
     * 代理类
     */
    private IVoucherOrderService proxy;

    /**
     * 一旦初始化就开始执行任务（Bean的后处理器）
     */
    @PostConstruct
    private void init() {
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }

    /**
     * 内部任务类，使用的Redis的消息队列（解决了异步秒杀中出现的内存问题）
     */
    private class VoucherOrderHandler implements Runnable {
        @Override
        public void run() {
            while(true) {
                try {
                    if(stringRedisTemplate == null) {
                        return;
                    }
                    // 1. 获取消息队列中的订单信息
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create("stream.orders", ReadOffset.lastConsumed())
                    );

                    // 2. 判断消息是否获取成功
                    if(list == null || list.isEmpty() ) {
                        // 2.1 如果获取消息失败说明没有消息，继续下一次循环
                        continue;
                    }

                    // 2.2 有消息，可以下单
                    // 解析消息中的订单
                    MapRecord<String, Object, Object> entries = list.get(0);
                    Map<Object, Object> value = entries.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);

                    // 3. 创建订单（创建订单中仍然要判断是否已经下单）
                    handleVoucherOrder(voucherOrder);

                    // 4. ack确认，告诉消息队列消息已经被处理了
                    stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", entries.getId());
                } catch (Exception e) {
                    log.info("处理异常，将会放入pending-list");
                    handlePendingList();
                }
            }
        }

        private void handlePendingList() {
            while(true) {
                try {
                    // 1. 获取pending队列中的订单信息
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1),
                            StreamOffset.create("stream.orders", ReadOffset.from("0"))
                    );

                    // 2. 判断消息是否获取成功
                    if(list == null || list.isEmpty() ) {
                        // 2.1 如果获取消息失败说明没有消息结束循环
                        break;
                    }

                    // 2.2 有消息，可以下单
                    // 解析消息中的订单
                    MapRecord<String, Object, Object> entries = list.get(0);
                    Map<Object, Object> value = entries.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);

                    // 3. 创建订单
                    handleVoucherOrder(voucherOrder);

                    // 4. ack确认，告诉消息队列消息已经被处理了
                    stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", entries.getId());

                    handleVoucherOrder(voucherOrder);
                } catch (Exception e) {
                   log.error("处理pending-list订单异常", e);
                   try {
                       Thread.sleep(200);
                   } catch (Exception ex) {
                       log.error("线程睡眠异常！");
                   }
                }
            }
        }
    }

    /**
     * 异步秒杀01（使用的是JVM阻塞队列），线程任务
     */
    private class VoucherOrderHandler01 implements Runnable {
        @Override
        public void run() {
            while(true) {
                try {
                    // 1. 获取队列中的订单信息
                    VoucherOrder voucherOrder = orderTasks.take();

                    // 2. 创建订单
                    handleVoucherOrder(voucherOrder);
                } catch (Exception e) {
                    log.error("处理订单异常:{}", e.getMessage());
                    throw new RuntimeException();
                }
            }
        }
    }

    /**
     * 创建订单，还可以二次加锁，更加安全（异步不会有性能影响）
     * @param voucherOrder
     */
    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        // 1. 获取用户，因为是从线程池中获取用户所以不能使用ThreadLocal
        Long userId = voucherOrder.getUserId();

        ///////////////  注意这儿可不使用锁（因为线程池是单线程 + Lua脚本的控制）   ////////////////
        // 2. 创建锁对象
        RLock lock = redissonClient.getLock("lock:order:" + userId);

        // 3. 获取锁 （如果是异步下单，则可不需要锁，因为线程池执行任务是单线程的）
        boolean isLock = lock.tryLock();

        // 4. 获取锁失败,返回错误信息，要么重试
        if(!isLock) {
            log.error("不允许重复下单");
            return;
        }

        try {
            this.proxy.createVoucherOrder(voucherOrder);
        } catch (Exception e) {
            log.error("发生错误:{}", e.getMessage());
        } finally {
            lock.unlock();
        }
    }

    /**
     * 异步秒杀使用Stream消息队列
     * @param voucherId     秒杀券Id
     * @return              秒杀券结果
     */
    @Override
    public Result seckillVoucher(Long voucherId) {
        // 1. 获取用户Id
        Long userId = UserHolder.getUser().getId();

        // 2. 获取订单Id（使用Redis生成ID保证了ID的唯一）
        long orderId = redisIdWorker.nextId("order");

        // 3. 执行lua脚本（将会判断是否有资格，比如用户是否买过，库存是否充足，我们保证在这个期间数据是不过期的）
        // 消息在Lua脚本中发送给了消息队列
        Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(), userId.toString(), String.valueOf(orderId));

        // 4. 判断结果是否为0
        int r;
        if((r = result.intValue()) != 0) {
            // 4.1 不为0代表没有购买资格
            return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
        }

        // 5. 返回订单id
        return Result.ok(orderId);
    }

    /////////////////// 异步秒杀01【使用阻塞队列】 ////////////////////
    public Result seckillVoucher05(Long voucherId) {
        // 1. 获取用户id
        Long id = UserHolder.getUser().getId();

        // 2. 执行lua脚本（将会判断是否有资格，比如用户是否买过，库存是否充足，我们保证在这个期间数据是不过期的）
        Long result = stringRedisTemplate.execute(SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(), id.toString());

        // 3. 判断结果是否为0
        int r;
        if((r = result.intValue()) != 0) {

            // 3.1 不为0代表没有购买资格
            return Result.fail(r == 1 ? "库存不足": " 不能重复下单");
        }

        // 4. 为0则有购买资格，保存到阻塞队列

        // 4.1 准备订单信息
        long orderId = redisIdWorker.nextId("order");
        VoucherOrder voucherOrder = new VoucherOrder();
        voucherOrder.setVoucherId(voucherId);
        voucherOrder.setUserId(UserHolder.getUser().getId());
        voucherOrder.setId(redisIdWorker.nextId("order:"));

        // 4.2 放入阻塞队列
        orderTasks.add(voucherOrder);

        // 5. 获取代理对象
        this.proxy = (IVoucherOrderService)AopContext.currentProxy();

        // 6. 返回订单id
        return Result.ok(orderId);
    }

    /////////////////// 使用[Redisson分布式锁]解决一人一单问题，使用乐观锁解决超卖问题 ////////////////////
    public Result seckillVoucher04(Long voucherId) {
        // 1. 根据优惠卷id查询优惠卷信息
        SeckillVoucher seckillVoucher = seckillVoucherMapper.selectById(voucherId);
        if(seckillVoucher == null) {
            throw new RuntimeException();
        }

        // 2. 判断秒杀是否开始，如果没开始返回异常
        if(seckillVoucher.getBeginTime().isAfter(LocalDateTime.now())) {
            return Result.fail("活动未开始！");
        }

        // 3. 判断秒杀是否结束，如果结束返回异常
        if(seckillVoucher.getEndTime().isBefore(LocalDateTime.now())) {
            return Result.fail("活动已结束！");
        }

        // 4. 判断库存是否充足，不充足返回异常
        if(seckillVoucher.getStock() < 1) {
            return Result.fail("已枪光，下次再来吧！");
        }

        // 5. 解决一人一单，对用户Id加锁
        Long userId = UserHolder.getUser().getId();

        // 6. 尝试创建锁对象
        RLock lock = redissonClient.getLock("lock:order:" + userId);

        // 7. 尝试获取锁(可自己确定重试和超时释放时间，这儿不给了)
        boolean isLock = lock.tryLock();

        // 8. 未获取锁则返回失败，不允许重试
        if(!isLock) {
            return Result.fail("不允许重复下单！");
        }

        // 9. 解决一人一单且防止超卖
        try {
            IVoucherOrderService proxy = (IVoucherOrderService)AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        } finally {
            lock.unlock();
        }
    }

    //////////////////// 使用[自定义分布式锁 + Lua脚本]解决集群一人一单问题，使用乐观锁解决超卖问题 ////////////////
    public Result seckillVoucher03(Long voucherId) {
        // 1. 根据优惠卷id查询优惠卷信息
        SeckillVoucher seckillVoucher = seckillVoucherMapper.selectById(voucherId);
        if(seckillVoucher == null) {
            throw new RuntimeException();
        }

        // 2. 判断秒杀是否开始，如果没开始返回异常
        if(seckillVoucher.getBeginTime().isAfter(LocalDateTime.now())) {
            return Result.fail("活动未开始！");
        }

        // 3. 判断秒杀是否结束，如果结束返回异常
        if(seckillVoucher.getEndTime().isBefore(LocalDateTime.now())) {
            return Result.fail("活动已结束！");
        }

        // 4. 判断库存是否充足，不充足返回异常
        if(seckillVoucher.getStock() < 1) {
            return Result.fail("已枪光，下次再来吧！");
        }

        // 5. 解决一人一单，对用户Id加锁
        Long userId = UserHolder.getUser().getId();

        // 6. 尝试创建锁对象
        SimpleRedisLock simpleRedisLock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);

        // 7. 尝试获取锁（并传递锁的超时释放时间，不是重试时间）
        boolean isLock = simpleRedisLock.tryLock(1200);

        // 8. 未获取锁则返回失败，不允许重试
        if(!isLock) {
            return Result.fail("不允许重复下单！");
        }

        // 9. 解决一人一单且防止超卖
        try {
            IVoucherOrderService proxy = (IVoucherOrderService)AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        } finally {
            simpleRedisLock.unlock();
        }
    }

    /////////////////// 使用悲观锁解决一人一单问题，使用乐观锁解决超卖问题  /////////////////////
    public Result seckillVoucher02(Long voucherId) {
        // 1. 根据优惠卷id查询优惠卷信息
        SeckillVoucher seckillVoucher = seckillVoucherMapper.selectById(voucherId);
        if(seckillVoucher == null) {
            throw new RuntimeException();
        }

        // 2. 判断秒杀是否开始，如果没开始返回异常
        if(seckillVoucher.getBeginTime().isAfter(LocalDateTime.now())) {
            return Result.fail("活动未开始！");
        }

        // 3. 判断秒杀是否结束，如果结束返回异常
        if(seckillVoucher.getEndTime().isBefore(LocalDateTime.now())) {
            return Result.fail("活动已结束！");
        }

        // 4. 判断库存是否充足，不充足返回异常
        if(seckillVoucher.getStock() < 1) {
            return Result.fail("已枪光，下次再来吧！");
        }

        // 5. 解决一人一单，对用户Id加锁(集群条件下会有问题)
        Long userId = UserHolder.getUser().getId();

        // 6. 对不同的userId进加锁
        synchronized (userId.toString().intern()) {

            // 7. 必须拿到代理对象，让代理对象调用，否则事务会失效（这儿需要添加依赖和注解）
            IVoucherOrderService proxy = (IVoucherOrderService)AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        }
    }

    ////////////////////////////// 仅仅使用CAS解决了超卖问题 ////////////////////////////
    public Result seckillVoucher01(Long voucherId) {
        // 1. 根据优惠卷id查询优惠卷信息
        SeckillVoucher seckillVoucher = seckillVoucherMapper.selectById(voucherId);

        // 2. 如果秒杀券不存在则抛出异常
        if(seckillVoucher == null) {
            throw new RuntimeException();
        }

        // 3. 判断秒杀是否开始，如果没开始返回异常
        if(seckillVoucher.getBeginTime().isAfter(LocalDateTime.now())) {
            return Result.fail("活动未开始！");
        }

        // 4. 判断秒杀是否结束，如果结束返回异常
        if(seckillVoucher.getEndTime().isBefore(LocalDateTime.now())) {
            return Result.fail("活动已结束！");
        }

        // 5. 判断库存是否充足，不充足返回异常
        if(seckillVoucher.getStock() < 1) {
            return Result.fail("已枪光，下次再来吧！");
        }

        // 6. 判断该用户是否已经下过单（存在线程安全，后面解决）
        Long userId = UserHolder.getUser().getId();
        QueryWrapper<VoucherOrder> voucherOrderQueryWrapper = new QueryWrapper<>();
        voucherOrderQueryWrapper.eq("user_id", userId);
        voucherOrderQueryWrapper.eq("voucher_id", voucherId);
        Integer cntRes = voucherOrderMapper.selectCount(voucherOrderQueryWrapper);

        // 7. 用户已经下单则返回异常
        if(cntRes > 0) {
            return Result.fail("抢票失败，一个用户只能抢一张！");
        }

        // 8. 如果库存充足且用户未下单则使用CAS， 以stock为不变量削减库存
        boolean isSuccess = seckillVoucherService.update().
                setSql("stock = stock - 1").eq("voucher_id", voucherId).gt("stock", 0).update();

        // 9. 只有CAS更新成功了才能执行下面的语句
        if(isSuccess) {

            // 10. 创建订单并插入数据库
            VoucherOrder voucherOrder = new VoucherOrder();
            voucherOrder.setVoucherId(voucherId);
            voucherOrder.setUserId(UserHolder.getUser().getId());
            voucherOrder.setId(redisIdWorker.nextId("order:"));
            voucherOrderMapper.insert(voucherOrder);
            return Result.ok(voucherOrder.getId());
        }

        // 11. 如果success未false则未抢到
        return Result.fail("未抢到！");
    }

    /**
     * 创建优惠券订单项
     * @param voucherId     优惠券Id
     * @return              创建优惠券的状态
     */
    @Transactional
    public  Result createVoucherOrder(Long voucherId) {
        // 1. 充足则判断用户是否下过单没有
        Long userId = UserHolder.getUser().getId();
        QueryWrapper<VoucherOrder> voucherOrderQueryWrapper = new QueryWrapper<>();
        voucherOrderQueryWrapper.eq("user_id", userId);
        voucherOrderQueryWrapper.eq("voucher_id", voucherId);
        Integer cntRes = voucherOrderMapper.selectCount(voucherOrderQueryWrapper);

        if(cntRes > 0) {
            // 2. 用户已经下过单
            return Result.fail("抢票失败，一个用户只能抢一张！");
        }

        // 3. 库存充足则扣减库存
        boolean isSuccess = seckillVoucherService.update().
                setSql("stock = stock - 1").eq("voucher_id", voucherId).gt("stock", 0).update();

        // 4. 只有更新成功了才能执行下面的语句
        if(isSuccess) {

            // 5. 创建订单
            VoucherOrder voucherOrder = new VoucherOrder();
            voucherOrder.setVoucherId(voucherId);
            voucherOrder.setUserId(UserHolder.getUser().getId());
            voucherOrder.setId(redisIdWorker.nextId("order:"));
            voucherOrderMapper.insert(voucherOrder);

            // 6. 返回订单id
            return Result.ok(voucherOrder.getId());
        }

        return Result.fail("未抢到！");
    }

    /**
     * 异步秒杀相关的创建订单（会再次检查数据库）
     * @param voucherOrder 订单信息
     */
    @Transactional
    public  void createVoucherOrder(VoucherOrder voucherOrder) {

        // 1. 充足则判断用户是否下过单没有 （可不用判断，因为Lua脚本已经做了）
        Long userId = voucherOrder.getUserId();
        QueryWrapper<VoucherOrder> voucherOrderQueryWrapper = new QueryWrapper<>();
        voucherOrderQueryWrapper.eq("user_id", userId);
        voucherOrderQueryWrapper.eq("voucher_id", voucherOrder.getVoucherId());
        Integer cntRes = voucherOrderMapper.selectCount(voucherOrderQueryWrapper);
        if(cntRes > 0) {
            log.error("抢票失败，一个用户只能抢一张！");
            return;
        }

        // 2. 库存充足则扣减库存
        boolean success = seckillVoucherService.update().
                setSql("stock = stock - 1").eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0).update();


        if(!success) {
            log.error("发生错误！");
            return;
        }

        // 3. 保存订单到数据库
        voucherOrderMapper.insert(voucherOrder);
    }
}
